package kafka

import com.mysql.cj.api.MysqlConnection
import com.mysql.cj.x.protobuf.MysqlxCrud

import java.sql.{PreparedStatement, ResultSet, Statement}

//import java.sql.{PreparedStatement, ResultSet, Statement}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe

import scala.util.parsing.json.JSON

object SparkStreamingKafka {
  def main(args: Array[String]): Unit = {
    Logger.getLogger("org").setLevel(Level.WARN)

    var conf = new SparkConf().setMaster("local[4]").setAppName("kafka")
    val ssc = new StreamingContext(conf, Seconds(5))
    ssc.checkpoint("d:\\streaming")

    var kafkaParams = Map[String, Object](
      elems = "bootstrap.servers" -> "master:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "test"
    )
    val topics = Array("danmaku_list")

    val kafkaStream = KafkaUtils.createDirectStream[String, String](ssc, PreferConsistent,
      Subscribe[String, String](topics, kafkaParams))

    var rs: ResultSet = null

    val ds = kafkaStream.flatMap(danmaku => JSON.parseFull(danmaku.value()).get.asInstanceOf[Map[String, Any]].get("content"))
                        .map(value => (value, 1))
                        .reduceByKey(_ + _)
                        .updateStateByKey((currentValue: Seq[Int], preValue: Option[Int]) => {
                          val currentCount = currentValue.sum
                          val preCount = preValue.getOrElse(0)
                          Some(currentCount + preCount)
                        }).foreachRDD((rdd, index) => {
                        println("-------------------------------------")
                        rdd.foreach(item => {
                          val SELECT = s"SELECT id FROM barrage_shield WHERE content = '${item._1}'"
                          rs = MySQLConnect.conn().createStatement().executeQuery(SELECT)
                          if (!rs.next()) {
                            println("not have")
                            MySQLConnect.conn().createStatement().executeUpdate(
                              s"INSERT INTO barrage_shield(content, times) VALUES('${item._1}',${item._2})"
                            )
                          } else {
                            println("has")
                            MySQLConnect.conn().createStatement().executeUpdate(
                              s"UPDATE barrage_shield SET times = ${item._2} WHERE content = '${item._1}'"
                            )
                          }
                          println(item._1 + ": " + item._2)
                        })
                      })
    ssc.start()
    ssc.awaitTermination()
  }
}